package org.fujene.communicate.repli;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Enumeration;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.fujene.index.ReplicateIndexSystem;
import org.fujene.memman.ReplicateMemoryManager;
import org.fujene.search.ReplicateSearchSystem;
import org.fujene.structure.Constants;
import org.fujene.structure.ContentInfo;
import org.fujene.structure.ExtendQueue;
import org.fujene.structure.ExtendResult;
import org.fujene.structure.ExtendTerm;
import org.fujene.structure.HashID;
import org.fujene.structure.LookupPointer;
import org.fujene.structure.NodeInfo;
import org.fujene.structure.SearchQueue;
import org.fujene.structure.Term;
import org.fujene.structure.TermList;
import org.fujene.term.Chopper;
import org.fujene.term.SenChopper;
import org.fujene.toolbox.ContentHash;
import org.msgpack.MessagePackObject;
import org.msgpack.object.ArrayType;
import org.msgpack.object.IntegerType;
import org.msgpack.object.RawType;
import org.msgpack.rpc.Client;
import org.msgpack.rpc.Server;
import org.msgpack.rpc.loop.EventLoop;

public class ReplicatePeerSystem {
	public static final int OUTER = 0;
	public static final int INNER = 1;

	// MemoryManager -> Memory, Disk
	private ReplicateMemoryManager memman = null;

	public boolean stopFlag = false;

	private ExecutorService execIndex = null;
	private ExecutorService execSearch = null;
	private ExecutorService execNodeLive = null;

	private String directory = null;
	private int blockNum = 0;
	private int blockSize = 0;

	// //////// P2P ///////////
	private static final int PEERPORT = Constants.PEER_PORT;

	private int replication;
	private int indexType;

	private TreeSet<NodeInfo> outerNodes = null;
	private TreeSet<NodeInfo> innerNodes = null;
	private NodeInfo myOuterNode = null;
	private NodeInfo myInnerNode = null;

	private String[] firstHost = null;

	private Server server = null;

	private Chopper chopper = null;

	private ContentInfo contentInfo = null;

	// //////// Logger ///////////
	private Logger mainLogger = null;

	public void initialize() throws IOException, InterruptedException {
		outerNodes = new TreeSet<NodeInfo>();
		innerNodes = new TreeSet<NodeInfo>();

		// 自分のIPアドレス取得
		Enumeration<NetworkInterface> enulfs = NetworkInterface
				.getNetworkInterfaces();
		if (!enulfs.hasMoreElements())
			return;
		NetworkInterface ni = enulfs.nextElement();

		Enumeration<InetAddress> enuAddr = ni.getInetAddresses();
		if (!enuAddr.hasMoreElements())
			return;
		InetAddress addrInfo = enuAddr.nextElement();

		int[] hash = calcHash(addrInfo);
		String address = addrInfo.getHostAddress();
		myOuterNode = new NodeInfo(address, hash[0]);
		outerNodes.add(myOuterNode);
		myInnerNode = new NodeInfo(address, hash[1]);
		innerNodes.add(myInnerNode);
		getInformation();

		if (contentInfo != null) {
			contentInfo.optimize();
		} else {
			contentInfo = new ContentInfo(new String[] { "Content" },
					new String[0]);
		}

		server = new Server(EventLoop.defaultEventLoop());
		server.serve(new ReplicatePeerHandler(mainLogger, this, contentInfo
				.getContentNum(), contentInfo.getSupplementNum()));
		server.listen(PEERPORT);

		new ReplicateSearchSystem(this, mainLogger);

		new ReplicateIndexSystem(this, contentInfo.getContentNum(),
				contentInfo.getSupplementNum(), mainLogger);

		memman = new ReplicateMemoryManager(blockSize, blockNum, directory,
				mainLogger, contentInfo);

		execIndex = Executors.newFixedThreadPool(1);

		execSearch = Executors.newFixedThreadPool(1);

		execNodeLive = Executors.newFixedThreadPool(2);

		chopper = new SenChopper();
		execNodeLive.execute(new SurvivingService(OUTER));
		execNodeLive.execute(new SurvivingService(INNER));
		if (firstHost != null) {
			registerNode();
			callForData(OUTER);
			callForData(INNER);
		}
	}

	private boolean getArgs(String[] args) {
		boolean isEnded = false;
		String settingFile;

		if (args.length > 2 || args.length == 1) {
			System.err
					.println("Usage: PeerSystem [--setting <SettingFile>] [--secondary <IPAddress ...>]");
			return isEnded;
		}
		// デフォルト値
		this.blockNum = 256;
		this.blockSize = 1048576;
		this.directory = "./fujene_data/";
		this.indexType = 1;
		this.replication = 1;
		this.mainLogger = Logger.getLogger("Fujene search engine");

		// ローカル
		ArrayList<String> contentName = new ArrayList<String>();
		ArrayList<String> supplementName = new ArrayList<String>();
		int count = 0;

		if (args.length > 1 && args[0].equals("--setting")) {
			settingFile = args[1];

			BufferedReader br = null;
			String input = null;
			try {
				br = new BufferedReader(new FileReader(new File(settingFile)));

				while ((input = br.readLine()) != null) {
					// "#"始まりはコメント行
					if (!input.startsWith("#")) {
						String[] settingArgs = input.split("=");
						if (settingArgs.length >= 2) {
							if (settingArgs[0].equalsIgnoreCase("BlockNum"))
								blockNum = Integer.parseInt(settingArgs[1]);
							else if (settingArgs[0]
									.equalsIgnoreCase("BlockSize"))
								blockSize = Integer.parseInt(settingArgs[1]);
							else if (settingArgs[0]
									.equalsIgnoreCase("Directory"))
								directory = settingArgs[1];
							else if (settingArgs[0]
									.equalsIgnoreCase("PeerMode")) {
								if (settingArgs[1].equalsIgnoreCase("on")) {
									replication = 3;
								}
							} else if (settingArgs[0]
									.equalsIgnoreCase("replication"))
								replication = Integer.parseInt(settingArgs[1]);
							else if (settingArgs[0]
									.equalsIgnoreCase("IndexMode"))
								indexType = Integer.parseInt(settingArgs[1]);
							else if (settingArgs[0]
									.equalsIgnoreCase("LogLevel")) {
								try {
									mainLogger.setLevel(Level
											.parse(settingArgs[1]));
								} catch (Exception e) {
									e.printStackTrace();
									mainLogger
											.severe("Specified log-level does not exist.");
								}
							} else if (settingArgs[0]
									.equalsIgnoreCase("Content")) {
								contentName.add(settingArgs[1]);
							} else if (settingArgs[0]
									.equalsIgnoreCase("Supplement")) {
								supplementName.add(settingArgs[1]);
							} else if (settingArgs[0]
									.equalsIgnoreCase("Connection")) {
								// この時点で紐付け情報を作る
								if (contentInfo == null)
									contentInfo = new ContentInfo(
											(String[]) contentName.toArray(),
											(String[]) supplementName.toArray());

								String[] values = settingArgs[1].split(",");
								int[] indices = new int[values.length];
								for (int i = 0; i < indices.length; i++)
									indices[i] = Integer.parseInt(values[i]);
								contentInfo.connectGraph(count++, indices);
							}
						}
					}

					isEnded = true;
				}
			} catch (FileNotFoundException e) {
				mainLogger
						.severe("Please input appropriate file. Search system will shutdown.");
				return isEnded;
			} catch (IOException e) {
			} finally {
				if (br != null) {
					try {
						br.close();
					} catch (IOException e) {
					}
				}
			}
		} else if (args.length == 0) {
			isEnded = true;
		}

		// セカンダリモード
		if (args.length > 1 && args[0].equals("--secondary")) {
			firstHost = new String[args.length - 1];
			for (int i = 1; i < args.length; i++) {
				firstHost[i - 1] = args[i];
			}
			isEnded = true;
		}

		return isEnded;
	}

	// ローカルで動かしたいときはこっち
	public static void main(String[] args) throws InterruptedException,
			IOException {
		ReplicatePeerSystem mainSystem = new ReplicatePeerSystem();

		if (!mainSystem.getArgs(args))
			return;

		mainSystem.initialize();

		mainSystem.console();
	}

	private void console() throws IOException {
		BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
		String arg;

		try {
			Thread.sleep(700);
		} catch (InterruptedException ignored) {
		}

		while (true) {
			System.out.print("Fujene > ");
			arg = br.readLine();

			String[] args = arg.split(" ");
			if (args.length > 0) {
				if (args[0].equals("index") || args[0].equals("i")) {
					if (args.length == 1)
						System.out
								.println("Usage: index ID <Contents> [<Supplements>]");
					else {
						if (args.length < contentInfo.getContentNum()
								+ contentInfo.getSupplementNum() + 2) {
							System.out.println("Input appropriate contents.");
						} else {
							setNewIndexExtent(new ExtendQueue(
									Arrays.copyOfRange(args, 1,
											1 + contentInfo.getContentNum()),
									Arrays.copyOfRange(
											args,
											1 + contentInfo.getContentNum(),
											1
													+ contentInfo
															.getContentNum()
													+ contentInfo
															.getSupplementNum()),
									Long.parseLong(args[1])));
						}
					}
				} else if (args[0].equals("search") || args[0].equals("s")) {
					if (args.length == 1)
						System.out.println("Specify query.");
					else {
						try {
							ExtendQueue queue;
							int i, j;
							ExtendResult result = setNewSearchExtent(new SearchQueue(
									args[1], args[2], 0, 100000));
							// ちょっとなら結果を出す
							if (result.getLength() <= 10) {
								for (i = 0; i < result.getLength(); i++) {
									System.out.println(i + ": ");
									queue = result.getQueue(i);
									for (j = 0; j < queue.contents.length; j++) {
										System.out.print(contentInfo
												.getContentName(j) + " / ");
										System.out.println(queue.contents[j]);
									}
								}
							}
						} catch (Exception ignored) {
						}
					}
				} else if (args[0].equals("searchspec") || args[0].equals("p")) {
					if (args.length != 4)
						System.out.println("Specify arguments.");
					else {
						try {
							ExtendResult result = setNewSearchExtent(new SearchQueue(
									args[1], args[2],
									Integer.parseInt(args[3]),
									Integer.parseInt(args[4])));
							int i, j;
							ExtendQueue queue;
							System.out.println("Result total: "
									+ result.getLength());
							if (result.getLength() <= 10) {
								for (i = 0; i < result.getLength(); i++) {
									System.out.println(i + ": ");
									queue = result.getQueue(i);
									for (j = 0; j < queue.contents.length; j++) {
										System.out.print(contentInfo
												.getContentName(j) + " / ");
										System.out.println(queue.contents[j]);
									}
								}
							}
						} catch (Exception ignored) {
							ignored.printStackTrace();
						}
					}
				} else if (args[0].equals("indexcount") || args[0].equals("c")) {
					System.out.println("Total count: " + memman.getCounter());
					System.gc();
				} else if (args[0].equals("querycount") || args[0].equals("u")) {
					System.out.println("Total count: "
							+ memman.getSearchcounter());
				} else if (args[0].equals("termcount") || args[0].equals("t")) {
					System.out.println("Term count: " + memman.getTerms());
				} else if (args[0].equals("skipcount") || args[0].equals("k")) {
					System.out.println("Skip ptr. count: "
							+ memman.getSkipPointerLen());
				} else if (args[0].equals("memory") || args[0].equals("m")) {
					memman.getMemoryList();
				} else if (args[0].equals("save") || args[0].equals("v")) {
					saveMemory();
				} else if (args[0].equals("read") || args[0].equals("r")) {
					readMemory();
				} else if (args[0].equals("quit") || args[0].equals("q")) {
					setStopper(true);
				} else {
					System.out
							.println("Command: <i>ndex <args>, <s>earch <args> type, searchs<p>ec <args> from length, <t>ermcount, index<c>ount, q<u>erycount, s<k>ipcount, <m>emory, <r>ead, sa<v>e, <q>uit");
				}
			}
		}
	}

	// 外部からJUnitとかで呼び出すときはこっち
	public void testMain(String[] args) throws InterruptedException,
			IOException {
		this.blockNum = 32;
		this.blockSize = 1048576;
		this.directory = "./testDir/";
		this.initialize();
	}

	// 全部の動作を止める操作
	public void setStopper(boolean isStop) {
		stopFlag = isStop;
		execIndex.shutdownNow();
		execSearch.shutdownNow();
		// 無理やりな気がする
		Runtime.getRuntime().exit(0);
	}

	public void setNewIndexExtent(ExtendQueue newIndex) {
		execIndex.execute(new PeerIndexContentsHost(newIndex));
		for (int i = 0; i < newIndex.contents.length; i++) {
			TermList terms = chopper.chop(newIndex.contents[i]);
			for (Term term : terms) {
				execIndex.execute(new PeerIndexTermHost(term.str(), newIndex.ID, i));
			}
		}
	}

	public class PeerIndexContentsHost implements Runnable {
		private ExtendQueue newIndex;

		public PeerIndexContentsHost(ExtendQueue index) {
			this.newIndex = index;
		}

		@Override
		public void run() {
			long startTime;
			long hash;
			NodeInfo nextPeer;
			Client client;

			try {
				hash = ContentHash.getContentHash(newIndex.ID);
				startTime = System.nanoTime();
				nextPeer = getNodeByHash(hash, innerNodes);
				mainLogger.finest("Index / Get node: "
						+ ((System.nanoTime() - startTime) / 1000.0)
						+ " microsec.");
				MessagePackObject[] contentObjs = new MessagePackObject[newIndex.content_types
						+ newIndex.supplement_types];
				for (int i = 0; i < newIndex.content_types; i++) {
					contentObjs[i] = RawType.create(newIndex.contents[i]);
				}
				for (int i = 0; i < newIndex.supplement_types; i++) {
					contentObjs[i + newIndex.content_types] = RawType
							.create(newIndex.supplements[i]);
				}

				for (int i = 0; i < (replication > innerNodes.size() ? innerNodes
						.size() : replication); i++) {
					if (!nextPeer.equals(myInnerNode)) {
						client = new Client(nextPeer.getAddr(), PEERPORT);
						startTime = System.nanoTime();

						client.callApply(
								"indexContent",
								new Object[] { ArrayType.create(contentObjs),
										IntegerType.create(hash),
										IntegerType.create(newIndex.ID) });
						mainLogger.finest("Index / Invoke to remote: "
								+ ((System.nanoTime() - startTime) / 1000.0)
								+ " microsec.");

						client.close();
					} else {
						startTime = System.nanoTime();
						for (int j = 0; j < newIndex.content_types; j++) {
							memman.storeContent(newIndex.contents[j],
									newIndex.ID, j);
						}
						for (int j = 0; j < newIndex.supplement_types; j++) {
							memman.storePureContent(newIndex.supplements[j],
									newIndex.ID, j);
						}
						mainLogger.finest("Index / Index content: "
								+ ((System.nanoTime() - startTime) / 1000.0)
								+ " microsec.");
					}
					nextPeer = getNext(nextPeer, innerNodes);
				}
			} catch (Exception e) {
			}
		}
	}

	public class PeerIndexContentsClient implements Runnable {
		private ExtendQueue newIndex;

		public PeerIndexContentsClient(ExtendQueue index) {
			this.newIndex = index;
		}

		@Override
		public void run() {
			for (int i = 0; i < newIndex.content_types; i++) {
				memman.storeContent(newIndex.contents[i], newIndex.ID, i);
			}
			for (int i = 0; i < newIndex.supplement_types; i++) {
				memman.storePureContent(newIndex.supplements[i], newIndex.ID, i);
			}
		}
	}

	public class PeerIndexTermHost implements Runnable {
		private String newTerm;
		private long termHash;
		private long ID;
		private int valueType;

		public PeerIndexTermHost(String newTerm, long ID, int valueType) {
			this.newTerm = newTerm;
			this.termHash = ContentHash.getTermHash(newTerm);
			this.ID = ID;
			this.valueType = valueType;
		}

		@Override
		public void run() {
			NodeInfo postPeer = getNodeByHash(termHash, outerNodes);

			Client client;

			for (int i = 0; i < (replication > outerNodes.size() ? outerNodes
					.size() : replication); i++) {
				if (postPeer.equals(myOuterNode)) {
					memman.storeTerm(newTerm, ID, 1, valueType);
				} else {
					try {
						client = new Client(postPeer.getAddr(), PEERPORT);
						client.setRequestTimeout(5);

						client.callApply(
								"indexTerm",
								new Object[] { RawType.create(newTerm),
										IntegerType.create(ID),
										IntegerType.create(valueType) });

						client.close();
					} catch (UnknownHostException e) {
					}
				}
				postPeer = getNext(postPeer, outerNodes);
			}
		}
	}

	public class PeerIndexTermClient implements Runnable {
		private String newTerm;
		private long ID;
		private int valueType;

		public PeerIndexTermClient(String newTerm, long ID, int valueType) {
			this.newTerm = newTerm;
			this.ID = ID;
			this.valueType = valueType;
		}

		@Override
		public void run() {
			memman.storeTerm(newTerm, ID, 1, valueType);
		}
	}

	private CountDownLatch searchGate;
	private LookupPointer[] resultPtrs;
	private long[] termHashes;
	private boolean isNotFound;

	// 大元のsearch窓口
	public ExtendResult setNewSearchExtent(SearchQueue newQuery)
			throws Exception {
		ExtendResult result = new ExtendResult(contentInfo);

		newQuery.typenum = contentInfo.getContentIndex(newQuery.typestr);

		TermList terms = chopper.chop(newQuery.query);
		searchGate = new CountDownLatch((terms.size() > 10 ? 10 : terms.size()));
		resultPtrs = new LookupPointer[(terms.size() > 10 ? 10 : terms.size())];
		termHashes = new long[(terms.size() > 10 ? 10 : terms.size())];
		isNotFound = false;
		long time = System.nanoTime();

		int count = 0;
		for (Term term : terms) {
			// 処理を投げる
			execSearch
					.execute(new PeerSearchHost(term.str(), count, newQuery.typenum));
			if (++count > 10)
				break;
		}
		mainLogger.finest("Search / Sending process: "
				+ ((System.nanoTime() - time) / 1000.0) + " microsec.");
		time = System.nanoTime();
		// 全部が検索し終わったらインターセクション
		searchGate.await();
		mainLogger.finest("Search / Search terms: "
				+ ((System.nanoTime() - time) / 1000.0) + " microsec.");

		if (!isNotFound) {
			try {
				time = System.nanoTime();
				// インターセクション
				ArrayList<HashID> hashes = intersection();
				mainLogger.finest("Search / Intersection: "
						+ ((System.nanoTime() - time) / 1000.0) + " microsec.");
				time = System.nanoTime();
				// 本文抽出
				result.setResult(getContents(hashes, newQuery.from,
						newQuery.length, 0));
				mainLogger.finest("Search / Getting content: "
						+ ((System.nanoTime() - time) / 1000.0) + " microsec.");
			} catch (Exception e) {
				e.printStackTrace();
			}
		} else {
			System.out.println("Not found.");
			result.clearResult();
		}

		return result;
	}

	// 検索サービス
	public class PeerSearchHost implements Runnable {
		private String newQuery = null;
		private int cnt;
		private int valueType;

		public PeerSearchHost(String newQuery, int count, int valueType) {
			this.newQuery = newQuery;
			this.cnt = count;
			this.valueType = valueType;
		}

		public void run() {
			try {
				long termHash = ContentHash.getTermHash(newQuery);
				termHashes[cnt] = termHash;
				NodeInfo termNode = getNodeByHash(termHash, outerNodes);

				mainLogger.fine("Search - Node: " + termNode.toString());
				if (termNode.equals(myOuterNode)) {
					resultPtrs[cnt] = memman.searchTerm(newQuery, valueType);
				} else {
					try {
						Client client = new Client(termNode.getAddr(), PEERPORT);

						MessagePackObject ret = client.callApply("searchTerm",
								new Object[] { newQuery, valueType });
						if (ret.isArrayType()) {
							MessagePackObject[] objs = ret.asArray();
							resultPtrs[cnt] = new LookupPointer(
									objs[0].asInt(), objs[1].asInt(),
									objs[2].asInt());
						}
					} catch (UnknownHostException e) {
					}
				}
				// 検索結果が出たらカウントダウン
				if (resultPtrs[cnt] == null)
					isNotFound = true;
			} finally {
				searchGate.countDown();
			}
		}
	}

	// 一度に読み込むHashIDの数
	private static final int BUFFERSIZE = 100;

	private ArrayList<Long> getContentList(long hash, LookupPointer fromPtr,
			int length) {
		ArrayList<Long> hashList = null;

		NodeInfo termPeer = getNodeByHash(hash, outerNodes);
		if (termPeer.equals(myOuterNode)) {
			hashList = memman.getContentsByHash(fromPtr, length);
		} else {
			try {
				Client client = new Client(termPeer.getAddr(), PEERPORT);
				client.setRequestTimeout(5);

				MessagePackObject obj = client.callApply(
						"getList",
						new Object[] { IntegerType.create(fromPtr.block),
								IntegerType.create(fromPtr.point),
								IntegerType.create(fromPtr.firstID),
								IntegerType.create(length) });
				if (obj.isArrayType()) {
					MessagePackObject[] objs = obj.asArray();
					if (objs.length > 1) {
						fromPtr.block = objs[0].asInt();
						fromPtr.point = objs[1].asInt();
						fromPtr.firstID = objs[2].asInt();
						hashList = new ArrayList<Long>((objs.length - 3) + 1);
						for (int i = 3; i < objs.length; i++) {
							hashList.add(Long.valueOf(objs[i].asLong()));
						}
					}
				}
			} catch (UnknownHostException e) {
				e.printStackTrace();
			}
		}

		return hashList;
	}

	private ArrayList<HashID> intersection() {
		ArrayList<HashID> results = new ArrayList<HashID>();

		int ptrlen = resultPtrs.length;
		long[] currentPtrs = new long[ptrlen];
		@SuppressWarnings("rawtypes")
		ArrayList[] buffer = new ArrayList[ptrlen];
		int[] currentNum = new int[ptrlen];
		int counter = 0;
		boolean samePtr = false;
		boolean isDone = false;
		int newestIndex;
		long newestPtr;
		long comps;
		int i;

		// 計算量: O(n)
		for (i = 0; i < ptrlen; i++) {
			buffer[i] = getContentList(termHashes[i], resultPtrs[i], BUFFERSIZE);
			currentPtrs[i] = ((Long) buffer[i].get(0)).longValue();
			currentNum[i] = 0;
		}

		while (counter < 100000) {
			samePtr = true;
			newestIndex = 0;
			newestPtr = currentPtrs[0];
			// ポインタの比較
			for (i = 1; i < ptrlen; i++) {
				comps = newestPtr - currentPtrs[i];
				if (comps < 0) {
					newestIndex = i;
					newestPtr = currentPtrs[i];
					samePtr = false;
				} else if (samePtr && comps > 0) {
					samePtr = false;
				}
			}

			if (samePtr) {
				// 一致 全ポインタ進め
				results.add(new HashID(ContentHash.getContentHash(newestPtr),
						newestPtr));
				counter++;
				for (i = 0; i < ptrlen; i++) {
					// もうない
					if (++currentNum[i] >= BUFFERSIZE) {
						if (resultPtrs[i].equals(Constants.emptyPointer)) {
							// もうない
							isDone = true;
							break;
						}
						// 新しいバッファを取る
						buffer[i] = getContentList(termHashes[i],
								resultPtrs[i], BUFFERSIZE);
						currentNum[i] = 0;
					}
					if (currentNum[i] >= buffer[i].size()) {
						isDone = true;
						break;
					}
					currentPtrs[i] = ((Long) buffer[i].get(currentNum[i]))
							.longValue();
				}
			} else {
				// 一番新しいものを進める

				currentNum[newestIndex]++;
				if (currentNum[newestIndex] >= BUFFERSIZE) {
					if (resultPtrs[newestIndex].equals(Constants.emptyPointer)) {
						// これ以上揃いようがない
						isDone = true;
						break;
					}
					buffer[newestIndex] = getContentList(
							termHashes[newestIndex], resultPtrs[newestIndex],
							BUFFERSIZE);
					currentNum[newestIndex] = 0;
				}
				// もうない
				if (currentNum[newestIndex] >= buffer[newestIndex].size()) {
					isDone = true;
					break;
				}
				currentPtrs[newestIndex] = ((Long) buffer[newestIndex]
						.get(currentNum[newestIndex])).longValue();
			}

			if (isDone)
				break;
		}

		System.out.println("Result count: " + results.size());

		return results;
	}

	public ExtendQueue[] getContents(ArrayList<HashID> hash, int from,
			int length, int priorIndex) {
		ExtendQueue[] result = new ExtendQueue[(length > hash.size() ? hash
				.size() : length)];
		NodeInfo lookupPeer;
		Client client;
		int count = 0;
		int i;

		for (HashID eachHash : hash) {
			if (count >= from && count < from + length) {
				lookupPeer = getNodeByHash(eachHash.hash, innerNodes);
				if (lookupPeer.equals(myInnerNode)) {
					result[count] = memman.lookupContent(eachHash.ID,
							priorIndex);
				} else {
					try {
						client = new Client(lookupPeer.getAddr(), PEERPORT);
						MessagePackObject obj = client.callApply("getContent",
								new Object[] { eachHash.ID });
						if (obj.isArrayType()) {
							MessagePackObject[] array = obj.asArray();
							String[] peerContent = new String[contentInfo
									.getContentNum()];
							String[] peerSupplement = new String[contentInfo
									.getSupplementNum()];

							for (i = 0; i < peerContent.length; i++) {
								peerContent[i] = array[i].asString();
							}
							for (i = 0; i < peerSupplement.length; i++) {
								peerSupplement[i] = array[i
										+ peerContent.length].asString();
							}
						}
					} catch (UnknownHostException e) {
					}
				}
			}
			count++;
		}

		return result;
	}

	public void readMemory() {
		memman.initializeMemory();
	}

	public void saveMemory() {
		memman.saveMemory();
	}

	// ///////////////////// P2P関連の実装 //////////////////////
	// ノードを返します
	public TreeSet<NodeInfo> getNode(int type) {
		if (type == OUTER)
			return outerNodes;
		else if (type == INNER)
			return innerNodes;
		else
			return null;
	}

	// インデックスを投げます
	public void invokeIndexInner(ExtendQueue queue) {
		execIndex.execute(new PeerIndexContentsClient(queue));
	}

	// インデックスを投げます
	public void invokeIndexOuter(String str, long ID, int valueType) {
		execIndex.execute(new PeerIndexTermClient(str, ID, valueType));
	}

	public LookupPointer invokeSearch(String query, int valueType) {
		return memman.searchTerm(query, valueType);
	}

	public ArrayList<ExtendQueue> getContentListByHash(int hashFrom, int hashTo) {
		return memman.getContentList(hashFrom, hashTo, 0);
	}

	public ArrayList<ExtendTerm> getTermListByHash(int hashFrom, int hashTo,
			int valueType) {
		return memman.getTermList(hashFrom, hashTo, valueType);
	}

	public ArrayList<Long> getContentsByHash(LookupPointer ptr, int length) {
		return memman.getContentsByHash(ptr, length);
	}

	public ArrayType getInfo() {
		return ArrayType
				.create(new MessagePackObject[] { IntegerType.create(blockNum),
						IntegerType.create(blockSize),
						IntegerType.create(replication),
						IntegerType.create(indexType) });
	}

	public int getQueueSize(int type) {
		if (type == 0)
			return ((ThreadPoolExecutor) execIndex).getPoolSize();
		else if (type == 1)
			return ((ThreadPoolExecutor) execSearch).getPoolSize();
		else
			return -1;
	}

	public ExtendQueue lookupContent(long ID, int index) {
		return memman.lookupContent(ID, index);
	}

	// 自分のハッシュ値を計算する
	private int[] calcHash(InetAddress addrInfo) {
		// OuterとInner用に
		int[] returnHash = new int[2];

		// とりあえずマシン名で作る
		returnHash[0] = addrInfo.getHostName().hashCode() * 0x2468ACE1;
		Integer.rotateLeft(returnHash[1], 17);
		// とりあえずかき混ぜる
		returnHash[1] = returnHash[0] ^ 0xC46F1043;
		Integer.rotateLeft(returnHash[1], 11);

		return returnHash;
	}

	// n個前(hashが小さい方)のピアを取得
	private NodeInfo getBackHash(NodeInfo from, int length,
			TreeSet<NodeInfo> circle) {
		if (from == null)
			return null;
		NodeInfo currentPeer = from;

		for (int i = 0; i < length; i++) {
			currentPeer = circle.lower(currentPeer);
			if (currentPeer == null) {
				// ループ
				currentPeer = circle.last();
			}
		}

		return currentPeer;
	}

	// n個前(hashが小さい方)のピアを取得
	private NodeInfo getFrontHash(NodeInfo from, int length,
			TreeSet<NodeInfo> circle) {
		if (from == null)
			return null;
		NodeInfo currentPeer = from;

		for (int i = 0; i < length; i++) {
			currentPeer = circle.higher(currentPeer);
			if (currentPeer == null) {
				// ループ
				currentPeer = circle.first();
			}
		}

		return currentPeer;
	}

	// 次のピアを取得
	private NodeInfo getNext(NodeInfo from, TreeSet<NodeInfo> circle) {
		return getFrontHash(from, 1, circle);
	}

	// 前のピアを取得
	private NodeInfo getPrev(NodeInfo from, TreeSet<NodeInfo> circle) {
		return getBackHash(from, 1, circle);
	}

	// ハッシュ直後のピアを取得
	private NodeInfo getNodeByHash(long hash, TreeSet<NodeInfo> circle) {
		NodeInfo currentPeer = new NodeInfo("Dummy", hash);

		if (circle.ceiling(currentPeer) != null)
			return circle.ceiling(currentPeer);
		else
			return circle.first();
	}

	// ピア間がサークル内でいくつ離れているか取得 0はエラー
	private int getDistance(NodeInfo from, NodeInfo to, TreeSet<NodeInfo> circle) {
		// 2つのピアがあるか確認
		if (!circle.contains(from) || !circle.contains(to))
			return 0;

		NodeInfo nextPeer = getNext(from, circle);
		NodeInfo prevPeer = getPrev(from, circle);
		for (int i = 1;; i++) {
			if (nextPeer.equals(to))
				return i;
			if (prevPeer.equals(to))
				return (-1 * i);
			nextPeer = getNext(nextPeer, circle);
			prevPeer = getPrev(prevPeer, circle);
		}
	}

	// サークル情報を取得します。
	private void getInformation() {
		if (firstHost != null) {
			Client client = null;
			for (String hosts : firstHost) {
				try {
					client = new Client(hosts, PEERPORT);
					client.setRequestTimeout(3);

					// 情報取得
					MessagePackObject infoobj = client.callApply("getInfo",
							new Object[] {});
					MessagePackObject[] infos;
					if (infoobj.isArrayType()) {
						infos = infoobj.asArray();
						blockNum = infos[0].asInt();
						blockSize = infos[1].asInt();
						replication = infos[2].asInt();
						indexType = infos[3].asInt();
					} else
						return;

					break;
				} catch (UnknownHostException e) {
					continue;
				}
			}
		}
	}

	// 追加操作を行います。
	public void addReplication(int type, NodeInfo newPeer) {
		// サークルに追加
		if (type == OUTER)
			outerNodes.add(newPeer);
		else if (type == INNER)
			innerNodes.add(newPeer);
	}

	// ノードを登録します。
	private void registerNode() {
		MessagePackObject circle;
		Client client;
		MessagePackObject[] objs;

		for (String host : firstHost) {
			// ノードリスト要求
			try {
				client = new Client(host, PEERPORT);
				client.setRequestTimeout(5);

				circle = client.callApply("circleList", new Object[] { OUTER });
				if (circle.isArrayType()) {
					objs = circle.asArray();
					for (int i = 0; i < objs.length; i += 2)
						outerNodes.add(new NodeInfo(objs[i].asString(),
								objs[i + 1].asInt()));
				} else {
					client.close();
					continue;
				}

				circle = client.callApply("circleList", new Object[] { INNER });
				if (circle.isArrayType()) {
					objs = circle.asArray();
					for (int i = 0; i < objs.length; i += 2)
						innerNodes.add(new NodeInfo(objs[i].asString(),
								objs[i + 1].asInt()));
				} else {
					// やり直し
					client.close();
					outerNodes.clear();
					continue;
				}
			} catch (UnknownHostException e) {
				continue;
			}

			// 現時点でのノードリストを使って自分のことを送信
			TreeSet<NodeInfo> currentNode = new TreeSet<NodeInfo>(outerNodes);
			Client nodeClient;
			for (NodeInfo eachNode : currentNode) {
				if (!eachNode.equals(myOuterNode)) {
					try {
						nodeClient = new Client(eachNode.getAddr(), PEERPORT);
						nodeClient.setRequestTimeout(5);

						nodeClient.callApply(
								"register",
								new Object[] {
										RawType.create(myOuterNode.getAddr()),
										IntegerType.create(myOuterNode
												.getHash()),
										IntegerType.create(myInnerNode
												.getHash()), });

						nodeClient.close();
					} catch (UnknownHostException e) {
					}
				}
			}
			break;
		}
	}

	// データ要求をします。
	private void callForData(int type) {
		TreeSet<NodeInfo> nodes;
		long hashFrom, hashTo;
		int indexNum = 0;
		NodeInfo myNode;
		MessagePackObject[] objs;
		Client client = null;

		if (type == OUTER) {
			nodes = outerNodes;
			myNode = myOuterNode;
		} else {
			nodes = innerNodes;
			myNode = myInnerNode;
		}

		// ハッシュ範囲の計算
		if (nodes.size() > replication) {
			hashFrom = getBackHash(myNode, replication, nodes).getHash();
			hashTo = myNode.getHash();
		} else {
			// ぜんぶ
			hashFrom = 0;
			hashTo = 0xFFFFFFFF;
		}

		NodeInfo nextNode = getNext(myNode, nodes);

		try {
			client = new Client(nextNode.getAddr(), PEERPORT);
			client.setRequestTimeout(50);

			MessagePackObject obj = client.callApply(
					"getDataSize",
					new Object[] { IntegerType.create(type),
							IntegerType.create(hashFrom),
							IntegerType.create(hashTo),
							RawType.create(myNode.getAddr()) });

			if (obj.isIntegerType()) {
				indexNum = obj.asInt();
				System.out.println("Read " + indexNum + " records.");
			}

			for (int i = 0; i < indexNum; i++) {
				obj = client.callApply(
						"getData",
						new Object[] { IntegerType.create(i),
								RawType.create(myNode.getAddr()) });

				if (obj.isArrayType()) {
					objs = obj.asArray();

					if (type == OUTER) {
						// 転置インデックスも全部追加
						ArrayList<Long> invIndexList = new ArrayList<Long>();
						LookupPointer pointer = new LookupPointer(
								objs[3].asInt(), objs[4].asInt(),
								objs[2].asLong());

						while (!pointer.equals(Constants.emptyPointer)) {
							// TODO: 受け取るデータをなんとかする！
							MessagePackObject hashObj = client
									.callApply(
											"getList",
											new Object[] {
													IntegerType
															.create(pointer.block),
													IntegerType
															.create(pointer.point),
													IntegerType
															.create(pointer.firstID),
													IntegerType
															.create(BUFFERSIZE) });
							if (hashObj.isArrayType()) {
								MessagePackObject[] hashObjs = hashObj
										.asArray();
								if (hashObjs.length > 1) {
									pointer.block = hashObjs[0].asInt();
									pointer.point = hashObjs[1].asInt();
									pointer.firstID = hashObjs[2].asLong();
									for (int j = 3; j < hashObjs.length; j++) {
										invIndexList.add(Long
												.valueOf(hashObjs[j].asLong()));
									}
								}
							}
						}

						for (Long val : invIndexList) {
							execIndex.execute(new PeerIndexTermClient(objs[0]
									.asString(), val.longValue(), 0));
						}
					} else {
						long ID = objs[0].asLong();
						String[] contents = new String[contentInfo
								.getContentNum()];
						String[] supplements = new String[contentInfo
								.getSupplementNum()];
						for (int j = 0; j < contents.length; j++) {
							contents[j] = objs[j + 1].asString();
						}
						for (int j = 0; j < supplements.length; j++) {
							contents[j] = objs[j + 1 + contents.length]
									.asString();
						}
						execIndex.execute(new PeerIndexContentsClient(
								new ExtendQueue(contents, supplements, ID)));
					}
				}
			}

			client.callApply("endDataSend",
					new Object[] { RawType.create(myNode.getAddr()) });

			client.close();
		} catch (UnknownHostException e) {
		}
	}

	// 削除操作を行います。
	public void deleteReplication(int type, NodeInfo delPeer) {
		NodeInfo myNode;
		TreeSet<NodeInfo> circle;
		if (type == OUTER) {
			circle = outerNodes;
			myNode = myOuterNode;
		} else {
			circle = innerNodes;
			myNode = myInnerNode;
		}
		int dist = getDistance(myNode, delPeer, circle);
		if (circle.size() > replication && dist >= 1 && dist <= replication) {
			// レプリケーション維持に必要な範囲を決定
			NodeInfo dataPeer;
			long hashTo;

			// 削除ピアをよけてコピーできるピアを分散させる
			if (dist == replication) {
				dataPeer = getNext(myNode, circle);
				hashTo = delPeer.getHash();
			} else {
				dataPeer = getBackHash(myNode, replication - 1, circle);
				hashTo = dataPeer.getHash();
			}
			long hashFrom = getPrev(dataPeer, circle).getHash();

			// サークルから削除
			circle.remove(delPeer);

			// TODO: 繋げなかったときの対応
			try {
				Client client = new Client(dataPeer.getAddr(), PEERPORT);
				client.setRequestTimeout(5);

				int indexNum = 0;
				MessagePackObject[] objs;

				MessagePackObject obj = client.callApply(
						"getDataSize",
						new Object[] { IntegerType.create(type),
								IntegerType.create(hashFrom),
								IntegerType.create(hashTo),
								RawType.create(myNode.getAddr()) });

				if (obj.isIntegerType()) {
					indexNum = obj.asInt();
					System.out.println("Read " + indexNum + " records.");
				}

				for (int i = 0; i < indexNum; i++) {
					obj = client.callApply(
							"getData",
							new Object[] { IntegerType.create(i),
									RawType.create(myNode.getAddr()) });

					if (obj.isArrayType()) {
						objs = obj.asArray();

						if (type == OUTER) {
							// 転置インデックスも全部追加
							ArrayList<Long> invIndexList = new ArrayList<Long>();
							LookupPointer pointer = new LookupPointer(
									objs[3].asInt(), objs[4].asInt(),
									objs[2].asLong());

							while (!pointer.equals(Constants.emptyPointer)) {
								// TODO: 受け取るデータをなんとかする！
								MessagePackObject hashObj = client
										.callApply(
												"getList",
												new Object[] {
														IntegerType
																.create(pointer.block),
														IntegerType
																.create(pointer.point),
														IntegerType
																.create(pointer.firstID),
														IntegerType
																.create(BUFFERSIZE) });
								if (hashObj.isArrayType()) {
									MessagePackObject[] hashObjs = hashObj
											.asArray();
									if (hashObjs.length > 1) {
										pointer.block = hashObjs[0].asInt();
										pointer.point = hashObjs[1].asInt();
										pointer.firstID = hashObjs[2].asLong();
										for (int j = 3; j < hashObjs.length; j++) {
											invIndexList.add(Long
													.valueOf(hashObjs[j]
															.asLong()));
										}
									}
								}
							}

							for (Long val : invIndexList) {
								execIndex
										.execute(new PeerIndexTermClient(
												objs[0].asString(), val
														.longValue(), 0));
							}
						} else {
							long ID = objs[0].asLong();
							String[] contents = new String[contentInfo
									.getContentNum()];
							String[] supplements = new String[contentInfo
									.getSupplementNum()];
							for (int j = 0; j < contents.length; j++) {
								contents[j] = objs[j + 1].asString();
							}
							for (int j = 0; j < supplements.length; j++) {
								contents[j] = objs[j + 1 + contents.length]
										.asString();
							}
							execIndex
									.execute(new PeerIndexContentsClient(
											new ExtendQueue(contents,
													supplements, ID)));
						}
					}
				}

				client.close();
			} catch (UnknownHostException e) {
			}
		} else {
			// サークルから削除
			circle.remove(delPeer);
		}
	}

	/**
	 * ピアの生存確認をします。
	 */
	public class SurvivingService implements Runnable {
		private int type;

		public SurvivingService(int type) {
			this.type = type;
		}

		@Override
		public void run() {
			while (true) {
				TreeSet<NodeInfo> infos = new TreeSet<NodeInfo>(
						type == OUTER ? outerNodes : innerNodes);
				Client client;
				MessagePackObject result;

				for (NodeInfo eachInfo : infos) {
					if ((type == OUTER && !eachInfo.equals(myOuterNode))
							|| (type == INNER && !eachInfo.equals(myInnerNode))) {
						try {
							client = new Client(eachInfo.getAddr(), PEERPORT);
							// 3秒だけ待ってやる
							client.setRequestTimeout(3);

							result = client
									.callApply("isLive", new Object[] {});
							if (!result.isRawType()) {
								// クライアント死亡と推定
								deleteReplication(type, eachInfo);
							} else if (stopFlag) {
								// 自分が死ぬことを通知
								client.callApply("delete",
										new Object[] { OUTER });
								client.callApply("delete",
										new Object[] { INNER });
							}
						} catch (Exception e) {
							mainLogger.info("Node " + eachInfo.getAddr()
									+ " seems to be dead.");
							// クライアント死亡と推定
							deleteReplication(type, eachInfo);
						}
					}
				}

				try {
					if (stopFlag) {
						break;
					}
					TimeUnit.SECONDS.sleep(60);
				} catch (Exception e) {
					continue;
				}
			}
		}
	}
}
